package cacheserial;

import lombok.extern.slf4j.Slf4j;
import com.chtwm.pool.QbdPoolConfigurable;

import java.util.concurrent.PriorityBlockingQueue;

/**
 * 如果队列中有任务则执行，没有任务就阻塞
 *
 * @author Aimy
 */
@Slf4j
public class CacheSerialQueueMonitor {

    /**
     * 读取任务的线程数
     */
    private Integer monitorThreadSize;

    /**
     * 每一个队列的长度
     */
    private Integer queueSize;

    public CacheSerialQueueMonitor(Integer monitorThreadSize, Integer queueSize) {
        this.monitorThreadSize = monitorThreadSize;
        this.queueSize = queueSize;
    }

    /**
     * 开始
     */
    public void start() throws Exception {

        log.info("缓存串行化监听任务启动：ThreadPoolSIze:{},MonitorThreadSize:{},QueueSize:{}",
                QbdPoolConfigurable.getPoolSize(), monitorThreadSize, queueSize);

        //监听线程要常驻内存，不能被挂起，所以不能大于线程池的线程数
        if (monitorThreadSize > QbdPoolConfigurable.getPoolSize()) {
            String msg = String.format("监听线程数-%d-超出线程池的配置-%d！",
                    monitorThreadSize, queueSize, QbdPoolConfigurable.getPoolSize());
            log.error(msg);
            throw new Exception(msg);
        }

        //每一个监听线程都对应一个队列
        for (int i = 0; i < monitorThreadSize; ++i) {
            //新建一个监听队列,相同的key，写在前，读在后
            PriorityBlockingQueue<ICacheSerialRequest> queue =
                    new PriorityBlockingQueue<>(queueSize, (o1, o2) -> {
                        if (o2.getKey().equals(o1.getKey())) {
                            return o2.getType().compareTo(o1.getType());
                        } else {
                            return 0;
                        }
                    });
            //加入到指定位置，方便共享使用
            CacheSerialQueue.addQueue(queue);
            //启动一个线程监听该队列
            QbdPoolConfigurable.submit(() -> {
                while (true) {
                    log.info("从任务队列中取值...");
                    ICacheSerialRequest request = queue.take();
                    log.info("获取到任务 - {}，开始执行...", request.getDsc());
                    request.processor();
                }
            });
        }

    }
}
